package com.example.rabbitmqretry.service;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.TimeUnit;

//消费异常之后的重复消息，并保证幂等性
//注意：如果设置为并发消费，还需要对加锁来保证幂等性
@Component
@Slf4j
public class RabbitReceiver {

    //幂等性判断的前缀
    public static final String REPEAT_CHECK = "repeat_check:";
    //重试次数的前缀
    public static final String RETRY_COUNT = "retry_count:";
    //最大重试次数
    public static final int MAX_RETRY_COUNT = 3;
    @Autowired
    RedisTemplate<String,Object> redisTemplate;

    // 监听队列默认是单线程的，也可以设置 concurrency 来设置多线程，如果配置了 concurrency 为多线程要考虑幂等判断的并发问题
    @RabbitListener(queues = "${mq.queueBinding.queue}")
    public void payReceive(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        Map<String, Object> headers = message.getMessageProperties().getHeaders();
        String msgId = (String) headers.get("msgId");
        log.info("receive msgId {}",msgId);
        log.info("接受的消息：{}",msg);

        // 1.从redis中判断重复消费标记 msgId 是否存在(幂等判断)
        // 2.如果存在则说明已经处理过了，手动ack 并返回
        Boolean isExists = redisTemplate.hasKey(REPEAT_CHECK + msgId);
        if (isExists != null && isExists){
            log.info("{} 消息已经被消费",msgId);
            channel.basicAck(deliveryTag,false);
            return;
        }
        // 3.如果没处理过则继续执行，如果没有出现异常，则在redis中保存重复消费标记 msgId，然后直接手动ack
        try {
            // 处理业务逻辑，模拟异常
            int x = 1/0;
            redisTemplate.opsForValue().set(REPEAT_CHECK + msgId,msgId,60, TimeUnit.MINUTES);
            channel.basicAck(deliveryTag,false);
        } catch (Exception e) {
            log.error("消息消费失败：{}",e.getMessage());
            // 4.在执行过程中出现异常，删除重复消费标记 msgId，以免失败重试时，被误认为是重复消息
            redisTemplate.delete(REPEAT_CHECK + msgId);
            // 5.如果超过最大重试次数，则拒绝消息，进入死信队列，并从redis删除重试次数标记
            Integer retryCount = (Integer) redisTemplate.opsForValue().get(RETRY_COUNT + msgId);
            log.info("重试次数：{}",retryCount);
            if (retryCount != null && retryCount >= MAX_RETRY_COUNT){
                channel.basicReject(deliveryTag,false);
                redisTemplate.delete(RETRY_COUNT + msgId);
                return;
            }else {
                // 6.在redis中保存已重试的次数，如果为空，则初始化为1，否则+1，
                redisTemplate.opsForValue().increment(RETRY_COUNT + msgId);
                redisTemplate.expire(RETRY_COUNT + msgId,60,TimeUnit.MINUTES);
            }
            throw e;
        }

    }

}
